package com.bowlong.lang;

import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.bowlong.concurrent.async.Async;
import com.bowlong.sql.DAOAsync;
import com.bowlong.util.DateEx;
import com.bowlong.util.NewDate;

@SuppressWarnings({ "unchecked", "rawtypes" })
public class ThreadEx {
	private static final MyThreadFactory getThreadFactory(String name) {
		return new MyThreadFactory(name, true);
	}

	// 创建一个无限线程池
	static ThreadFactory cachedFactory;

	public static final ExecutorService newCachedThreadPool() {
		return newCachedThreadPool("cachedPool");
	}

	public static final ExecutorService newCachedThreadPool(String name) {
		if (cachedFactory == null) {
			cachedFactory = getThreadFactory(name);
		}
		ExecutorService r2 = Executors.newCachedThreadPool(cachedFactory);
		return r2;
	}

	// 创建一个固定大小线程
	static ThreadFactory fixedFactory;

	public static final ExecutorService newFixedThreadPool(int nThreads) {
		return newFixedThreadPool("fixedPool", nThreads);
	}

	public static final ExecutorService newFixedThreadPool(String name,
			int nThreads) {
		if (fixedFactory == null) {
			fixedFactory = getThreadFactory(name);
		}
		ExecutorService r2 = Executors.newFixedThreadPool(nThreads,
				fixedFactory);
		return r2;
	}

	// 单线程
	static ThreadFactory singleFactory;

	public static final ExecutorService newSingleThreadExecutor() {
		return newSingleThreadExecutor("singlePool");
	}

	public static final ExecutorService newSingleThreadExecutor(String name) {
		if (singleFactory == null)
			singleFactory = getThreadFactory(name);
		ExecutorService singlePool = Executors
				.newSingleThreadExecutor(singleFactory);
		return singlePool;
	}

	public static ExecutorService newThreadExecutor() {
		final int min = 0;
		final int max = Integer.MAX_VALUE;
		return newThreadExecutor(min, max);
	}

	public static ExecutorService newThreadExecutor(final int min, final int max) {
		return new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS,
				new ArrayBlockingQueue<Runnable>(max),
				new ThreadPoolExecutor.CallerRunsPolicy());
	}

	public static ExecutorService newThreadExecutor(final int min,
			final int max, final ThreadFactory threadFactory) {
		return new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS,
				new ArrayBlockingQueue<Runnable>(max), threadFactory,
				new ThreadPoolExecutor.CallerRunsPolicy());
	}

	// 用线程池执行任务
	public static final void execute(final Executor executor, final Runnable r) {
		executor.execute(r);
	}

	public static final Future submit(final ExecutorService executor,
			final Callable c) {
		return executor.submit(c);
	}

	public static final Thread newThreadExec(final Runnable r) {
		final Thread t = new Thread(r);
		t.start();
		return t;
	}

	public static final ForkJoinPool newForkJoinPool(int parallelism) {
		return new ForkJoinPool(parallelism);
	}

	public static final ForkJoinTask<?> forkSubmit(final ForkJoinPool fjp,
			final Runnable r) {
		return fjp.submit(r);
	}

	public static final ForkJoinTask<?> forkSubmit(final ForkJoinPool fjp,
			final Callable c) {
		return fjp.submit(c);
	}

	public static final ForkJoinTask<?> forkSubmit(final ForkJoinPool fjp,
			final RecursiveAction c) {
		return fjp.submit(c);
	}

	// /////////////////////////////////////////////////////////////////
	// 用线程池执行
	static ExecutorService _executor = null;

	public static final void execute(final Runnable r) {
		if (_executor == null)
			_executor = newCachedThreadPool();

		execute(_executor, r);
	}

	static ExecutorService _singleExecutor = null;

	public static final void executeSingle(final Runnable r) {
		if (_singleExecutor == null)
			_singleExecutor = newSingleThreadExecutor();

		execute(_singleExecutor, r);
	}

	static ThreadFactory scheduledFactory;

	public static final ScheduledExecutorService newScheduledPool(
			final int nThreads) {
		return newScheduledPool("scheduledPool", nThreads);
	}

	public static ScheduledExecutorService newScheduledPool(final String name,
			final int nThreads) {
		if (scheduledFactory == null)
			scheduledFactory = getThreadFactory(name);
		return Executors.newScheduledThreadPool(nThreads, scheduledFactory);
	}

	// /////////////////////////////////////////////////////////////////
	// 使用调度器执行
	static ScheduledExecutorService _scheduledPool = null;

	// 定时执行
	public static final ScheduledFuture<?> schedule(
			final ScheduledExecutorService ses, final Runnable r,
			final long delay, final TimeUnit unit) {
		return ses.schedule(r, delay, unit);
	}

	public static final ScheduledFuture<?> schedule(
			ScheduledExecutorService ses, Callable r, long delay, TimeUnit unit) {
		return ses.schedule(r, delay, unit);
	}

	public static final ScheduledFuture<?> schedule(
			final ScheduledExecutorService ses, final Runnable r,
			final long delay) {
		return ses.schedule(r, delay, TimeUnit.MILLISECONDS);
	}

	public static final ScheduledFuture<?> schedule(final Runnable r,
			final long delay) {
		return schedule(r, delay, TimeUnit.MILLISECONDS);
	}

	public static final ScheduledFuture<?> schedule(final Runnable r,
			final long delay, final TimeUnit unit) {
		if (_scheduledPool == null)
			_scheduledPool = newScheduledPool(32);

		return schedule(_scheduledPool, r, delay, unit);
	}

	public static final ScheduledFuture<?> schedule(final Callable r,
			final long delay) {
		return schedule(r, delay, TimeUnit.MILLISECONDS);
	}

	public static final ScheduledFuture<?> schedule(final Callable r,
			final long delay, final TimeUnit unit) {
		if (_scheduledPool == null)
			_scheduledPool = newScheduledPool(32);

		return schedule(_scheduledPool, r, delay, unit);
	}

	public static final FutureTask<Exception> execute(final ExecutorService es,
			final FutureTask<Exception> task) {
		es.submit(task);
		return task;
	}

	public static final FutureTask<Exception> execute(final ExecutorService es,
			final Callable<Exception> task) {
		FutureTask<Exception> ft = new FutureTask<Exception>(task);
		return execute(es, ft);
	}

	// 定时执行,再间隔调度重复执行
	public static final ScheduledFuture<?> scheduleWithFixedDelay(
			final ScheduledExecutorService scheduledPool, final Runnable r,
			final long initialDelay, final long delay, final TimeUnit unit) {
		return scheduledPool.scheduleWithFixedDelay(r, initialDelay, delay,
				unit);
	}

	public static final ScheduledFuture<?> scheduleWithFixedDelay(
			final ScheduledExecutorService scheduledPool, final Runnable r,
			final long initialDelay, final long delay) {
		return scheduledPool.scheduleWithFixedDelay(r, initialDelay, delay,
				TimeUnit.MILLISECONDS);
	}

	public static final ScheduledFuture<?> scheduleWithFixedDelay(
			final Runnable r, final long initialDelay, final long delay) {
		return scheduleWithFixedDelay(r, initialDelay, delay,
				TimeUnit.MILLISECONDS);
	}

	public static final ScheduledFuture<?> scheduleWithFixedDelay(
			final Runnable r, final long initialDelay, final long delay,
			final TimeUnit unit) {
		if (_scheduledPool == null)
			_scheduledPool = newScheduledPool(32);

		return scheduleWithFixedDelay(_scheduledPool, r, initialDelay, delay,
				unit);
	}

	public static final long now() {
		return System.currentTimeMillis();
	}

	// 定时执行
	public static final ScheduledFuture<?> execute(
			ScheduledExecutorService ses, final Runnable r, final Date d) {
		long delay = d.getTime() - now();
		delay = delay <= 0 ? 1 : delay;
		return ses.schedule(r, delay, TimeUnit.MILLISECONDS);
	}

	public static final ScheduledFuture<?> scheduled1(final Runnable r,
			final Date d) {
		if (_scheduledPool == null)
			_scheduledPool = Executors.newScheduledThreadPool(1);
		long delay = d.getTime() - now();
		delay = delay <= 0 ? 1 : delay;
		return _scheduledPool.schedule(r, delay, TimeUnit.MILLISECONDS);
	}

	// 确定时分秒，每日执行
	public static final ScheduledFuture<?> scheduled64EveryDay(
			final Runnable r, final int hour, final int minute, final int sec) {
		if (_scheduledPool == null)
			_scheduledPool = Executors.newScheduledThreadPool(32);

		NewDate dat = new NewDate();
		long now = now();
		long h = dat.getHour();
		long m = dat.getMinute();
		long s = dat.getSecound();
		long tomorrow = dat.addDay(-1).setHourMinuteSec(0, 0, 0).getTime();
		long initialDelay = 0;
		long e1 = h * DateEx.TIME_HOUR + m * DateEx.TIME_MINUTE + s
				* DateEx.TIME_SECOND;
		long e2 = hour * DateEx.TIME_HOUR + minute * DateEx.TIME_MINUTE + sec
				* DateEx.TIME_SECOND;
		if (e1 < e2) {
			initialDelay = e2 - e1;
		} else {
			initialDelay = tomorrow - now + e2;
		}
		long delay = DateEx.TIME_DAY;
		return _scheduledPool.scheduleAtFixedRate(r, initialDelay, delay,
				TimeUnit.MILLISECONDS);

	}

	public static final ScheduledFuture<?> scheduledEveryDay(
			ScheduledExecutorService SES, final Runnable r, final int hour,
			final int minute, final int sec) {

		NewDate dat = new NewDate();
		long now = now();
		long h = dat.getHour();
		long m = dat.getMinute();
		long s = dat.getSecound();
		long tomorrow = dat.addDay(-1).setHourMinuteSec(0, 0, 0).getTime();
		long initialDelay = 0;
		long e1 = h * DateEx.TIME_HOUR + m * DateEx.TIME_MINUTE + s
				* DateEx.TIME_SECOND;
		long e2 = hour * DateEx.TIME_HOUR + minute * DateEx.TIME_MINUTE + sec
				* DateEx.TIME_SECOND;
		if (e1 < e2) {
			initialDelay = e2 - e1;
		} else {
			initialDelay = tomorrow - now + e2;
		}
		long delay = DateEx.TIME_DAY;
		return SES.scheduleAtFixedRate(r, initialDelay, delay,
				TimeUnit.MILLISECONDS);
	}

	// /////////////////////////////////////////////////////////////////
	public static final void Sleep(final long t) {
		try {
			Thread.sleep(t);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void shutdown() {
		if (_executor != null) {
			_executor.shutdown();
			_executor = null;
		}
		if (_scheduledPool != null) {
			_scheduledPool.shutdown();
			_scheduledPool = null;
		}
		if (_singleExecutor != null) {
			_scheduledPool.shutdown();
			_scheduledPool = null;
		}
		DAOAsync.shutdown();
		Async.shutdown();
	}

	public static void shutdownNow() {
		if (_executor != null) {
			_executor.shutdownNow();
			_executor = null;
		}
		if (_scheduledPool != null) {
			_scheduledPool.shutdownNow();
			_scheduledPool = null;
		}
		if (_singleExecutor != null) {
			_scheduledPool.shutdownNow();
			_scheduledPool = null;
		}
		DAOAsync.shutdownNow();
		Async.shutdownNow();
	}

	// /////////////////////////////////////////////////////////////////

	// private class T implements Runnable {
	// String s;
	//
	// public T(String s) {
	// this.s = s;
	// }
	//
	// public void run() {
	// System.out.println(s + ":" + now());
	// }
	// }
	//
	// public void test() {
	// // 定时执行
	// ScheduledFuture f = schedule(new T("s"), 1000);
	// System.out.println(f.isDone());
	// // 定时间隔执行
	// ScheduledFuture f2 = scheduleWithFixedDelay(new T("swfd"), 1000, 1000);
	// Sleep(5000);
	// System.out.println(f.isDone());
	// f2.cancel(true);
	// Sleep(5000);
	// }
	//
	// public static void main(String[] args) {
	// ThreadEx t = new ThreadEx();
	// t.test();
	// }
}
